package com.rabbit.producer.broker;

import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
import com.google.common.collect.Maps;
import com.rabbit.api.Message;
import com.rabbit.api.MessageType;
import com.rabbit.api.exception.MessageRunTimeException;
import com.rabbit.common.convert.GenericMessageConverter;
import com.rabbit.common.convert.RabbitMessageConverter;
import com.rabbit.common.serializer.Serializer;
import com.rabbit.common.serializer.SerializerFactory;
import com.rabbit.common.serializerimpl.JacksonSerializerFactory;
import com.rabbit.producer.service.MessageStoreService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.Optional;

/**
 * RabbitTemplate线程池化
 * 每个topic 对应一个RabbitTemplate
 * 1.提高发送的效率
 * 2.可以根据不同的需求指定化不同的RabbitTemplate，每一个topic都有自己的routingKey规则
 *
 * @author: 顾志杰
 */
@Slf4j
@Component
public class RabbitTemplateContainer implements RabbitTemplate.ConfirmCallback{

    /**
     * Map<topic,RabbitTemplate>
     */
    private Map<String, RabbitTemplate> rabbitMap = Maps.newConcurrentMap();

    private Splitter splitter = Splitter.on("#");

    private SerializerFactory serializerFactory = JacksonSerializerFactory.INSTANCE;

    @Autowired
    private ConnectionFactory connectionFactory;

    @Autowired
    private MessageStoreService messageStoreService;


    public RabbitTemplate getRabbitTemplate(Message message)throws MessageRunTimeException {
        Preconditions.checkNotNull(message.getTopic());
        String topic = message.getTopic();
        RabbitTemplate rabbitTemplate = rabbitMap.get(topic);


        return Optional.ofNullable(rabbitTemplate).orElseGet(()->{
            log.info("#RabbitTemplateContainer.getRabbitTemplate# topic: {} is not exists, create one",topic);
            RabbitTemplate newRabbitTemplate = new RabbitTemplate(connectionFactory);
            newRabbitTemplate.setExchange(topic);
            newRabbitTemplate.setRetryTemplate(new RetryTemplate());
            newRabbitTemplate.setRoutingKey(message.getRoutingKey());
            /*对于message序列化*/
            /*添加序列化与反序列化和converter对象*/
            Serializer serializer = serializerFactory.create();
            GenericMessageConverter gmc = new GenericMessageConverter(serializer);
            RabbitMessageConverter rmc = new RabbitMessageConverter(gmc);
            newRabbitTemplate.setMessageConverter(rmc);

            String messageType = message.getMessageType();
            /*如果不是快速消息 都需要加上回调函数*/
            if(!MessageType.RAPID.equals(messageType)){
                newRabbitTemplate.setConfirmCallback(this);
            }
            rabbitMap.putIfAbsent(topic,newRabbitTemplate);

            return rabbitMap.get(topic);
        });
    }

    /**
     * 无论是 confirm 消息 还是 reliant 消息，发送消息以后broker都会去回调confirm
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        /* 具体的消息应答*/
        List<String> strings = splitter.splitToList(correlationData.getId());

        String messageId = strings.get(0);
        Long sendTime = Long.parseLong(strings.get(1));
        String messageType = strings.get(2);
        if(ack){
            /*当broker返回ack成功时，更新下日志表对应消息的status状态为SEND_OK*/

            /*如果当前消息类型为reliant 我们就去数据库查找并进行更新*/
            if(MessageType.RELIANT.endsWith(messageType)){
                this.messageStoreService.succuess(messageId);
            }
            log.info("send message is ok,confirm messageId:{},sendTime:{}",messageId,sendTime);
        }else{

            log.error("send message is fail,confirm messageId:{},sendTime:{}",messageId,sendTime);
        }

    }
}

